;; This file is part of GNUnet.
;; Copyright (C) 2012-2019, 2021 GNUnet e.V.
;;
;; GNUnet is free software: you can redistribute it and/or modify it
;; under the terms of the GNU Affero General Public License as published
;; by the Free Software Foundation, either version 3 of the License,
;; or (at your option) any later version.
;;
;; GNUnet is distributed in the hope that it will be useful, but
;; WITHOUT ANY WARRANTY; without even the implied warranty of
;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
;; Affero General Public License for more details.
;;
;; You should have received a copy of the GNU Affero General Public License
;; along with this program.  If not, see <http://www.gnu.org/licenses/>.
;;
;; SPDX-License-Identifier: AGPL-3.0-or-later

;; Author: Florian Dold
;; Author: Maxime Devos
;; C file: util/mq.c
;; Scheme module: (gnu gnunet mq envelope)
;;
;; Limitation: the format of messages is still in flux,
;; so no type checks there.
(define-library (gnu gnunet mq envelope)
  (export <envelope> make-envelope envelope?
	  attempt-cancel! attempt-irrevocable-sent!
	  envelope-peek-cancelled?
	  ;; TODO find a better place
	  (rename (bind-atomic-boxen %%bind-atomic-boxen)))
  (import (gnu gnunet utils hat-let)
	  (only (guile) define* lambda* exact-integer?)
	  (only (ice-9 match) match)
	  (only (ice-9 atomic)
		make-atomic-box atomic-box-ref
		atomic-box-compare-and-swap!)
	  (only (rnrs base)
		lambda assert letrec let begin define
		syntax-rules let-syntax define-syntax
		procedure? eq? >= = <= < if quote ...
		identifier-syntax values and let*
		vector vector-ref vector? vector-length)
	  (only (rnrs records syntactic) define-record-type))
  (begin
    (define-record-type (<envelope> make-envelope envelope?)
      ;; Atomic box:
      ;;    #t: cancelled
      ;;    #f: too late to cancel, message has been irrevocably sent!
      ;;
      ;;        (Unless you play tricks like pulling out the Ethernet
      ;;        cable before the message is received by the router)
      ;;    #(message prio notify-sent! cancel!)
      (fields (immutable state %cancellation-state))
      (protocol
       (lambda (%make)
	 (lambda* (cancel! message #:key (priority 0) (notify-sent! values))
	   "Make a message envelope; i.e., a record containing the message
(@var{message}, @var{priority}) and information on how to cancel the sending
of the message (@var{cancel!}) and who should be notified when the message
cannot be unsent anymore (@var{notify-sent!}).

Once marked as cancelled or irrevocably sent, the record drops its
references to @var{message}, @var{cancel!} and @var{notify-sent!}.
When being marked as cancelled, the thunk @var{cancel!} is called."
	   (assert (and (procedure? cancel!) (procedure? notify-sent!)
			(exact-integer? priority)
			(<= 0 priority 511)))
	   (%make (make-atomic-box
		   (vector message priority notify-sent! cancel!)))))))

    (define (envelope-peek-cancelled? envelope)
      "Test whether @var{envelope} is currently cancelled (true / false)."
      (eq? #t (atomic-box-ref (%cancellation-state envelope))))

    (define (%attempt-irrevocable-sent! envelope already-sent go cancelled)
      (bind-atomic-boxen
       ((state (%cancellation-state envelope) swap!))
       (let spin ((old state))
	 (match old
	   ;; See comment at %attempt-cancel! for
	   ;; why we don't do #(message prio notify-sent! cancel!)
	   ((? vector?)
	    (if (eq? old (swap! old #f))
		(let^ ((!! (= (vector-length old) 4))
		       (! message (vector-ref old 0))
		       (! prio (vector-ref old 1))
		       (! notify-sent! (vector-ref old 2)))
		      (notify-sent!)
		      (go message prio))
		(spin state)))
	   (#t (cancelled))
	   (#f (already-sent))))))

    (define-syntax attempt-irrevocable-sent!
      (syntax-rules (go cancelled already-sent)
	"If @var{envelope} is not cancelled and has not yet been sent,
mark the message as irrevocably sent, call the notify-sent callback and
evaluate @var{exp/go} in an environment where the message @var{message}
and its priority @var{priority} are bound.

If the message has already been marked as irrevocably sent,
evaluate @var{exp/already-sent} instead. If the message is cancelled,
evaluate @var{exp/cancelled} instead.

Even if this operation (and perhaps @code{attempt-cancel!}) is used concurrently
on the same @var{envelope}, whether by multi-threading, asynchronicities
(via @code{system-async-mark}) or by recursion, the following properties hold:

@begin itemize
@item the notify-sent callback of @var{envelope} is called at most once
@item the notify-sent callback is never called if @var{envelope} is cancelled
  at any point in time
@item likewise, the code in @var{exp/go} is at most evaluated once
@end itemize"
	((_ envelope
	    ((go message priority) . exp/go)
	    ((cancelled) . exp/cancelled)
	    ((already-sent) . exp/already-sent))
	 (%attempt-irrevocable-sent! envelope
				     (lambda () . exp/already-sent)
				     (lambda (message priority) . exp/go)
				     (lambda () . exp/cancelled)))))

    (define (%attempt-cancel! envelope now-cancelled already-cancelled
			      already-sent)
      (bind-atomic-boxen
       ((state (%cancellation-state envelope) swap!))
       (let spin ((old state))
	 (match old
	   ;; Do _not_ use #(message prio notify-sent! cancel!)
	   ;; here! Instead, delay the bounds check and accessing
	   ;; the elements of the vector after the swap!. That way:
	   ;;
	   ;; Premature optimisation.
	   ;;   We save a little time in case two threads try to concurrently
	   ;;   @var{state}.
	   ;;
	   ;; Meager excuse: self-healing (in case of memory corruption).
	   ;;   Suppose a cosmic ray flipped a few bits and now
	   ;;   @var{state} contains another vector, of different length.
	   ;;   Then by performing the swap before the bounds check,
	   ;;   the envelope is brought into a valid state. (And an
	   ;;   exception will still result.)
	   ((? vector?)
	    (if (eq? old (swap! old #t))
		(let^ ((!! (= (vector-length old) 4))
		       (! cancel! (vector-ref old 3)))
		      (cancel!)
		      (now-cancelled))
		(spin state)))
	   (#t (already-cancelled))
	   ;; XXX maybe make the meager excuse less meager
	   ;; and add a 'default' case where @var{state} is
	   ;; set to #f when bad (and an exception is raised)?
	   ;; Seems like some dedicated exception types for
	   ;; memory corruption are required then ...
	   ;; And tests.
	   (#f (already-sent))))))

    (define-syntax attempt-cancel!
      (syntax-rules (now-cancelled already-cancelled already-sent)
	"If @var{envelope} is not yet marked as cancelled or sent,
mark it as cancelled, call the corresponding cancellation callback
and evaluate @var{exp/now-cancelled}.

If @var{envelope} is already marked as cancelled, do not mutate
anything or call any callback and evaluate @var{exp/already-cancelled}.
Likewise, if @var{envelope} is marked as irrevocably sent, evaluate
@var{exp/already-sent} instead.

If this operation is interrupted before @var{exp/now-cancelled} is
evaluated, the envelope may be marked as cancelled even if the
cancellation callback has not yet been called or has not yet returned.

However, by tolerating this limitation, it can be (and is) guaranteed
that the cancellation callback is called at most once. Likewise, the
code in @var{exp/now-cancelled} is only be called at most once.
Also, the cancellation callback and is never called (and @var{exp/now-cancelled}
never evaluated) if @var{envelope} is marked as sent at any point in time."
	((_ envelope
	    ((now-cancelled) . exp/now-cancelled)
	    ((already-cancelled) . exp/already-cancelled)
	    ((already-sent) . exp/already-sent))
	 (%attempt-cancel! envelope
			   (lambda () . exp/now-cancelled)
			   (lambda () . exp/already-cancelled)
			   (lambda () . exp/already-sent)))))

    (define-syntax bind-atomic-boxen
      (syntax-rules ()
	((_ () exp exp* ...)
	 (let () exp exp* ...))
	((_ ((variable box swap!) . etc) exp exp* ...)
	 (let ((stashed-box box))
	   (let-syntax ((variable (identifier-syntax
				   (atomic-box-ref box))))
	     (let ((swap! (lambda (expected desired)
			    (atomic-box-compare-and-swap! box expected
							  desired))))
	       (bind-atomic-boxen etc exp exp* ...)))))))))
